package com.exd.flink.stream.function;

import java.util.Base64;
import java.util.Date;

import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import com.exceeddata.sdk.vdata.data.binary.BinaryLittleEndianUtils;
import com.exd.flink.common.DomainControlFlag;
import com.exd.flink.common.OutputVdataTagBase;
import com.exd.flink.common.StringTool;
import com.exd.flink.domain.VDataBean;
import com.exd.flink.domain.VDataErrorBean;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

/**
 * 解析转换VDATA的数据
 */
public class VdataAnalyticalFunction extends ProcessFunction<String, VDataBean> {

	private static final long serialVersionUID = 1L;

	@Override
	public void processElement(String message, Context context, Collector<VDataBean> collector) throws Exception {
		JsonElement msgElement = JsonParser.parseString(message);
		JsonObject jsonObject = msgElement.getAsJsonObject();
		// 车辆编号
		String vin = jsonObject.get("vin").getAsString();
		// 域控
		String serviceId = jsonObject.get("serviceId").getAsString();
		// 获取数据时间（检索使用）
		String year = jsonObject.get("year").getAsString();
		String month = jsonObject.get("month").getAsString();
		String day = jsonObject.get("day").getAsString();
		String hour = jsonObject.get("hour").getAsString();
		String min = jsonObject.get("min").getAsString();
		String sec = jsonObject.get("sec").getAsString();
		Long methodId = jsonObject.get("methodId").getAsLong();
		Long majorVersion = jsonObject.get("majorVersion").getAsLong();
		Long samplingTime = jsonObject.get("samplingTime").getAsLong();
		Long tboxSendTimestamp = jsonObject.get("tboxSendTimestamp").getAsLong();
		Long receiveTimestamp = jsonObject.get("receiveTimestamp").getAsLong();
		// 获取data数据
		String data = jsonObject.get("data").getAsString();
		try {
			String collectStringTime = year + "-" + month + "-" + day + " " + hour + ":" + min + ":" + sec;
			Date collectDateTime = StringTool.StringToDate(collectStringTime);
			// 解密data数据
			byte[] packetWithMetadata = Base64.getDecoder().decode(data.substring(0, 10));
			// 序号，同一个包的标识（不同的车辆有可能重复）
			int packetSequence = BinaryLittleEndianUtils.bytesToINT32(packetWithMetadata, 0);
			// 切片序号，当前是第几个切片
			int sliceNum = BinaryLittleEndianUtils.bytesToUINT16(packetWithMetadata, 4);
			// 是否是最后一个切片的标识 0：不是最后一个切片 1：最后一个切片
			int sliceLast = BinaryLittleEndianUtils.bytesToUINT8int(packetWithMetadata, 6);
			// 同一个包的唯一标识
			String key = serviceId + "_" + vin + "_" +collectDateTime.getTime()+ "_" + packetSequence;
			// 封装对象
			VDataBean vDataBean = new VDataBean();
			vDataBean.setVin(vin);
			vDataBean.setServiceId(serviceId);
			vDataBean.setCollectTime(collectDateTime.getTime());
			vDataBean.setData(data);
			vDataBean.setPacketSequence(packetSequence);
			vDataBean.setSliceNum(sliceNum);
			vDataBean.setSliceLast(sliceLast);
			vDataBean.setKey(key);
			vDataBean.setSamplingTime(samplingTime);
			vDataBean.setTboxSendTimestamp(tboxSendTimestamp);
			vDataBean.setReceiveTimestamp(receiveTimestamp);
			vDataBean.setYear(year);
			vDataBean.setMonth(month);
			vDataBean.setDay(day);
			vDataBean.setHour(hour);
			vDataBean.setMin(min);
			vDataBean.setSec(sec);
			vDataBean.setMethodId(methodId);
			vDataBean.setMajorVersion(majorVersion);
			// 输出正常的数据,根据不同的域控，输出到不同的流
			if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
				context.output(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_OK, vDataBean);
			} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
				context.output(OutputVdataTagBase.VDCM_ANALYTICAL_VDATA_OK, vDataBean);
			}
		} catch (Exception exception) {
			VDataErrorBean vDataErrorBean = new VDataErrorBean();
			vDataErrorBean.setServiceId(serviceId);
			vDataErrorBean.setVin(vin);
			vDataErrorBean.setYear(year);
			vDataErrorBean.setMonth(month);
			vDataErrorBean.setDay(day);
			vDataErrorBean.setHour(hour);
			vDataErrorBean.setMin(min);
			vDataErrorBean.setSec(sec);
			vDataErrorBean.setMethodId(methodId);
			vDataErrorBean.setMajorVersion(majorVersion);
			vDataErrorBean.setSamplingTime(samplingTime);
			vDataErrorBean.setTboxSendTimestamp(tboxSendTimestamp);
			vDataErrorBean.setReceiveTimestamp(receiveTimestamp);
			vDataErrorBean.setData(data);
			vDataErrorBean.setErrorType(0);
			vDataErrorBean.setErrorDescribe("数据解析异常");
			vDataErrorBean.setErrorMessage(exception.getMessage());
			vDataErrorBean.setCreateTime(new Date());
			// 输出异常的数据,根据不同的域控，输出到不同的流
			if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
				context.output(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_ERROR, vDataErrorBean);
			} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
				context.output(OutputVdataTagBase.VDCM_ANALYTICAL_VDATA_ERROR, vDataErrorBean);
			}
		}
	}

}
